package cn.pkubangbang.demo.democontact.service.impl.helper;

import java.time.Instant;
import java.util.Arrays;

/**
 * @Author: zhangzhenbang@k2data.com.cn
 * @Date: 2018-08-28 15:02
 */
public class Threshold {

    private long[] ring;
    private volatile long span;
    private int capacity;

    private volatile long recent;

    /**
     * the next index to insert data.
     */
    private volatile int head;

    /**
     * total count of the data in the ring.
     */
    private volatile int length;

    public Threshold() {
        this(200, 30000);
    }

    public Threshold(int maxHit, long timeSpanInMillis) {
        if (maxHit < 1) {
            throw new IllegalArgumentException("maxHit should be >= 1");
        }

        if (timeSpanInMillis < 50) {
            throw new IllegalArgumentException("timeSpanInMillis too small. should be >= 50 ms");
        }

        this.capacity = maxHit;
        this.ring = new long[capacity];
        this.span = timeSpanInMillis;
        this.head = 0;
        this.length = 0;
        this.recent = 0;
    }

    public long getSpan() {
        return span;
    }

    public void setSpan(long span) {
        this.span = span;
    }

    public long getRecent() {
        return recent;
    }

    public int getLength() {
        return length;
    }

    /**
     * push a new time into the ring.
     * @param newTime the new time
     * @return total count of data inside the ring
     * @throws ThresholdOverflowException if the ring is full and cannot hold the new data
     * @throws TimeRewindingException if the new time is of the past
     */
    public synchronized int push(long newTime) throws ThresholdOverflowException, TimeRewindingException {
        if (newTime < recent - 50) {
            // 50ms latency is allowed
            throw new TimeRewindingException("Passed in a time in the past! This ring is broken. Recent->"
                + Instant.ofEpochMilli(recent) + " Passed in->" + Instant.ofEpochMilli(newTime));
        }
        purge(newTime);
        if (length == capacity) {
            // full after purge, touch overflow
            throw new ThresholdOverflowException("HIT OVERFLOW! Reach "+ capacity +" hits in "+ span +" ms!");
        }
        recent = newTime;
        ring[head] = newTime;
        head = next(head);
        length++;
        return length;
    }

    /**
     * purge the ring against newTime
     * @param newTime the newTime. datum with time outside of the span get purged.
     * @return the count of the datum which get purged.
     */
    private synchronized int purge(long newTime) {
        int sum = 0;
        int count = length;
        if (count > 0) {
            int tail = getTail();
            do {
                if (ring[tail] + span < newTime) {
                    sum++;
                    count --;
                } else {
                    break;
                }
                tail = next(tail);
            } while (tail == head);
        }
        length = count;
        return sum;
    }

    public synchronized void clear() {
        this.ring = new long[capacity];
        this.head = 0;
        this.length = 0;
    }

    /**
     * Since the data is stored inside an 'array-like' ring, for next index, this method should be used.
     * @param curr current index
     * @return next index, ranging from 0~capacity
     */
    private int next(int curr) {
        return (++curr) % capacity;
    }

    /**
     * tail is calculated from head and length
     * @return the 'tail' position
     */
    private int getTail() {
        return (head + capacity - length) % capacity;
    }

    @Override
    public String toString() {
        final StringBuilder sb = new StringBuilder("Threshold{");
        sb.append("ring=").append(Arrays.toString(ring));
        sb.append(", span=").append(span);
        sb.append(", capacity=").append(capacity);
        sb.append(", recent=").append(recent);
        sb.append(", head=").append(head);
        sb.append(", length=").append(length);
        sb.append('}');
        return sb.toString();
    }
}
